package jobs

import (
	"encoding/json"
	"fmt"
	"gin-mvc/constants"
	"gin-mvc/dao"
	"gin-mvc/pkg/logging"
	"gin-mvc/pkg/setting"
	"github.com/gomodule/redigo/redis"
	"log"
	"strings"
	"time"
)

type JobMessage struct {
	Topic string
	Msg   string
	Data  interface{}
}

// Consumer 消费者
func Consumer() {
	if setting.JobSetting.Enable {
		log.Printf("[%s]队列开始消费...\n", setting.AppSetting.AppName)
		consumerStart()
	}
}

func consumerStart() {
	for i := 0; i < setting.JobSetting.ConsumerCount; i++ {
		go func() {
			for {
				jobs := GetJobs()
				if len(jobs) <= 0 {
					time.Sleep(time.Second)
					// fmt.Printf("延时队列无数据，等待1s\n")
					continue
				}
				for _, job := range jobs {
					// 必须消费掉，否则可能导致队列堆积，从而影响新任务的执行
					// 优化方案，设置堆积队列；取离当前30分钟的job进行自动消费；新增主动消费接口（未实现）
					if DeleteJob(job) > 0 {
						go handleMessage(job)
					}
				}
			}
		}()
	}
}

func handleMessage(job string) {
	jobMessage := JobMessage{}
	jobByte := []byte(job)
	if err := json.Unmarshal(jobByte, &jobMessage); err != nil {
		logging.Error(err)
		panic(err)
	}

	// 策略模式注册job处理逻辑
	switch jobMessage.Topic {
	case TestTopic:
		NewContext(NewTestJob()).Execute(&jobMessage)
	}
}

// generateJobKey 生成JobKey
func generateJobKey() string {
	return fmt.Sprintf(constants.JobMessageQueue, strings.ToUpper(setting.AppSetting.AppName))
}

// AddJob 添加任务
func AddJob(msg *JobMessage, delayTime int64) {
	conn := dao.RedisConn.Get()
	defer conn.Close()
	value, err := json.Marshal(msg)
	if err != nil {
		log.Fatal(err)
	}

	timeNow := time.Now().Unix()
	delayTime += timeNow

	if _, err := redis.Bool(conn.Do("ZADD", generateJobKey(), delayTime, string(value))); err != nil {
		panic("添加任务失败: " + err.Error())
	}
}

// GetJobs 获取任务
func GetJobs() []string {
	conn := dao.RedisConn.Get()
	defer conn.Close()
	timeNow := time.Now().Unix()
	ret, err := redis.Strings(conn.Do("zrangebyscore", generateJobKey(), 0, timeNow, "limit", 0, setting.JobSetting.JobCount))
	if err != nil {
		panic("获取Job失败：" + err.Error())
	}

	return ret
}

// DeleteJob 删除Job
func DeleteJob(value string) int {
	conn := dao.RedisConn.Get()
	defer conn.Close()
	res, err := redis.Int(conn.Do("zrem", generateJobKey(), value))
	if err != nil {
		panic("删除Job失败：" + err.Error())
	}

	return res
}
